package org.jetlinks.pro.things.impl.connector;

import com.alibaba.fastjson.JSON;
import lombok.*;
import org.apache.commons.codec.digest.DigestUtils;
import org.jetlinks.core.message.ThingMessage;
import org.jetlinks.core.things.ThingId;
import org.jetlinks.pro.streaming.Computer;
import org.jetlinks.pro.streaming.Streaming;
import org.jetlinks.pro.things.ThingConstants;
import org.jetlinks.pro.things.impl.connector.script.ScriptThingMessageStreaming;
import org.springframework.util.Assert;
import reactor.core.publisher.Flux;
import reactor.function.Function3;

import java.time.Duration;
import java.util.Collections;
import java.util.Set;
import java.util.function.BiFunction;
import java.util.function.Function;

@Getter
@Setter
@EqualsAndHashCode
public class ComputeRule {

    private WindowType windowType;

    private Window window;

    private String script;

    private AggType aggType = AggType.count;

    public void validate() {
        if (window != null) {
            window.validate();
            Assert.notNull(aggType, "[aggType]不能为空");
        }
        Assert.hasText(script, "[script]不能为空");
    }

    /**
     * 创建新的属性或事件计算
     *
     * @param resultConverter 计算结果转换器函数
     * @param <T> 消息计算结果对象
     */
    public <T> Streaming<ThingMessage, T, T> create(Function3<ThingId, Set<String>, Object, T> resultConverter) {

        //不支持窗口计算时，直接根据脚本规则和物ID构造计算流
        if (getWindowType() == null || getWindow() == null) {
            return ScriptThingMessageStreaming
                .create(script, (msg, result) -> resultConverter
                    .apply(ThingId.of(msg.getThingType(), msg.getThingId()),
                           msg.getHeaderOrElse(ThingConstants.connectorTrace, Collections::emptySet), result));
        }
        Computer<ThingMessage, Object> computer = ScriptThingMessageStreaming.create(script, (msg, result) -> result);

        Function<Flux<Object>, Flux<Flux<Object>>> windowFunction
            = (windowType == WindowType.time ? window::windowTime : window::window);
        //使用规则配置的md5值作为规则的唯一标示,避免重复重启相同的规则
        String jsonConfig = JSON.toJSONString(this);
        String identity = DigestUtils.md5Hex(jsonConfig);

        //根据脚本规则和物ID经过窗口计算返回计算流
        return ThingMessageComputeStreaming
            .create(identity,
                    computer,
                    flux -> windowFunction
                        .apply(flux)
                        .flatMap(window -> aggType
                                     .getComputer()
                                     .apply(window),
                                 Integer.MAX_VALUE),
                    resultConverter);
    }

    public enum WindowType {
        time,
        num
    }

    @Getter
    @Setter
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Window {

        //窗口长度
        private int span;

        //步长
        private int every;

        public void validate() {
            Assert.state(span > 0, "[span]不能小于等于0");
        }

        public Flux<Flux<Object>> window(Flux<Object> flux) {
            if (every <= 0) {
                return flux.window(span);
            }
            return flux
                .window(span, every);
        }

        public Flux<Flux<Object>> windowTime(Flux<Object> flux) {
            if (getEvery() <= 0) {
                return flux.window(Duration.ofSeconds(getSpan()));
            }
            return flux.window(Duration.ofSeconds(getSpan()), Duration.ofSeconds(getEvery()));
        }
    }


}